package com.lagou.client;

import com.lagou.RespSeq;
import com.lagou.coder.RpcEncoder;
import com.lagou.handler.UserClientHandler;
import com.lagou.protocol.RpcRequest;
import com.lagou.ser.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.AttributeKey;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEventType;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.logging.log4j.message.StringMapMessage;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * 消费者
 */
public class RPCConsumer {

    private static final String PATH = "/NETTY_IPS";

    private static final String RESP_SEQ = "/RESP_SEQ";

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    //2.声明一个自定义事件处理器  UserClientHandler
    private static Map<String, UserClientHandler> handlerMap = new ConcurrentHashMap<>();
    private static List<UserClientHandler> handlerList = new CopyOnWriteArrayList<UserClientHandler>();
    private static List<RespSeq> seqList = new CopyOnWriteArrayList<>();




    //3.编写方法,初始化客户端  ( 创建连接池  bootStrap  设置bootstrap  连接服务器)
    public static void initClient() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", new RetryNTimes(3, 1000));
        client.start();
        List<String> children = client.getChildren().forPath(PATH);
        for (String node : children){
            String[] ipPort = node.split(":");
            initNetty(ipPort[0], Integer.parseInt(ipPort[1]));

            if (client.checkExists().forPath(RESP_SEQ + "/" + node) == null){
                seqList.add(new RespSeq(node, 1));
            }else{
                String seq = new String(client.getData().forPath(RESP_SEQ + "/" + node));
                seqList.add(new RespSeq(node, Integer.parseInt(seq)));
            }

        }

        PathChildrenCache watcher = new PathChildrenCache(client, PATH, true);
        watcher.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event)
                    throws Exception {

                String dataPath = event.getData().getPath();
                PathChildrenCacheEvent.Type type = event.getType();
                if (dataPath.startsWith(PATH)){
                    if (type.equals(CuratorEventType.DELETE)){
                        String ipPort = new String(event.getData().getData());
                        handlerList.remove(handlerMap.get(ipPort));
                        handlerMap.remove(ipPort);
                        seqList = seqList.stream().filter(seq -> !ipPort.equals(seq.getIpPort())).collect(Collectors.toList());
                    }else if (type.equals(CuratorEventType.CREATE)) {
                        String data = new String(event.getData().getData());
                        String[] ipPort = data.split(":");
                        initNetty(ipPort[0], Integer.parseInt(ipPort[1]));
                        seqList.add(new RespSeq(data, 1));
                    }
                }
            }
        });
        watcher.start();

        PathChildrenCache watcherSeq = new PathChildrenCache(client, RESP_SEQ, true);
        watcherSeq.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event)
                    throws Exception {

                String dataPath = event.getData().getPath();
                PathChildrenCacheEvent.Type type = event.getType();
                if (dataPath.startsWith(RESP_SEQ)){
                    String[] split = dataPath.split("/");
                    String ipPort = split[split.length-1];
                    seqList = seqList.stream().filter(seq -> !ipPort.equals(seq.getIpPort())).collect(Collectors.toList());
                    if (!type.equals(CuratorEventType.DELETE)){
                        seqList.add(new RespSeq(ipPort, Integer.parseInt(new String(event.getData().getData()))));
                    }


                }
            }
        });
        watcherSeq.start();

    }

    private static void initNetty(String ip, Integer port) throws InterruptedException {
        //1) 初始化UserClientHandler
        UserClientHandler userClientHandler  = new UserClientHandler();

        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)创建客户端的引导对象
        Bootstrap bootstrap =  new Bootstrap();
        //4)配置启动引导对象
        bootstrap.group(group)
                //设置通道为NIO
                .channel(NioSocketChannel.class)
                //设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY,true)
                //监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //设置编码
//                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
//                        pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        //添加自定义事件处理器
                        pipeline.addLast(userClientHandler);
                    }
                });

        //5)连接服务端
        bootstrap.connect(ip,port).sync();
        handlerMap.put(ip+":"+port, userClientHandler);
        handlerList.add(userClientHandler);
    }

    //4.编写一个方法,使用JDK的动态代理创建对象
    // serviceClass 接口类型,根据哪个接口生成子类代理对象;   providerParam :  "UserService#sayHello#"
    public static Object createProxy(final Class<?> serviceClass){
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    public Object invoke(Object o, Method method, Object[] objects) throws Throwable {
                        //1)初始化客户端cliet
                        if(handlerMap.isEmpty()){
                            initClient();
                        }
                        seqList = seqList.stream().filter(seq -> StringUtils.isNotEmpty(seq.getIpPort())).collect(Collectors.toList());
                        System.out.println(seqList.toString());
                        String ipPort = null;
                        List<RespSeq> list = seqList.stream().filter(seq -> 1 != seq.getSeq().intValue()).collect(Collectors.toList());
                        if (list == null || list.size() == 0){
                            Collections.shuffle(seqList);
                            ipPort = seqList.get(0).getIpPort();
                        }else {
                            ipPort = seqList.stream().sorted(Comparator.comparingDouble(RespSeq::getSeq)).findFirst().get().getIpPort();
                        }
                        UserClientHandler userClientHandler = handlerMap.get(ipPort);

                        //2)给UserClientHandler 设置param参数
//                        userClientHandler.setParam("UserService#sayHello#"+objects[0]);
                        RpcRequest request = new RpcRequest();
                        request.setClassName(serviceClass.getName());
                        request.setMethodName(method.getName());
                        request.setParameterTypes(new Class[]{String.class});
                        request.setParameters(objects);
                        request.setRequestId(UUID.randomUUID().toString().replace("-",""));
//                        userClientHandler.setParam(JSON.toJSONString(request));
                        userClientHandler.setRequest(request);

                        //3).使用线程池,开启一个线程处理处理call() 写操作,并返回结果
                        Object result = executorService.submit(userClientHandler).get();


                        //4)return 结果
                        return result;
                    }
                });
    }

}
